package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"math/rand"
	"sort"

	//	"bytes"
	"sync"
	"sync/atomic"
	"time"

	"6.824/labgob"
	"6.824/labrpc"
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
	CommandTerm  int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//节点状态
type ServerState int

const (
	//领导者
	Leader ServerState = iota
	//追随者
	Follower
	//候选者
	Candidate
)

//日志条目
type LogEntry struct {
	//日志所属任期
	Term int
	//日志所属索引
	Index int
	//具体命令
	Command interface{}
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.RWMutex        // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	// 简化的日志对象，直接使用log.Logger
	logger *log.Logger // 日志记录器
	//提交信息的条件变量
	submitCond sync.Cond
	// 2A
	//更新日志和传消息的锁
	updateLock sync.Mutex
	//节点状态
	state ServerState
	//当前term
	CurrentTerm int
	//投票对象
	VotedFor int
	//日志条目（索引从1开始）
	Entries []LogEntry
	//选举超时timer
	electionTimer *time.Timer
	//心跳timer
	heartBeatTimer *time.Timer
	//2B
	//提交的日志的index，初始化为0，日志index从1开始
	commitIndex int
	//应用的日志的index，既提交给状态机的
	lastApplied int
	//要发给其他节点的日志的index，每次选举后初始化为leader日志条目+1
	nextIndex []int
	//已经复制的日志的最新条目。用于判断提交，选举后初始化为0
	matchIndex []int
	//上层消息传递通道
	applyChan chan ApplyMsg
	//消息分片发送队列，用来保证顺序性的
	applyQueue chan []ApplyMsg
	//传递commitIndex的通道
	commitIndexChan chan int
	//2D
	// 快照相关元数据
	LastIncludedIndex int // 快照包含的最后一条日志的索引
	LastIncludedTerm  int // 快照包含的最后一条日志的任期

	//用于保证快照信息和日志信息之间的同步性
	snapshotInProgress int        // 当前正在处理的快照索引（-1 表示无）
	snapshotCond       *sync.Cond // 条件变量
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.RLock()
	term = rf.CurrentTerm
	isleader = rf.state == Leader
	rf.mu.RUnlock()
	return term, isleader
}

//检查节点状态是否发生变化
func (rf *Raft) StateChanged(term int) bool {
	rf.mu.RLock()
	defer rf.mu.RUnlock()

	return rf.state != Leader || rf.CurrentTerm != term
}

func (rf *Raft) isLeader(term int) bool {
	rf.mu.RLock()
	defer rf.mu.RUnlock()

	return rf.state == Leader
}

//检查是否存在最新的任期的日志
func (rf *Raft) HasNoCurTermLog() bool {
	rf.mu.RLock()
	defer rf.mu.RUnlock()
	if len(rf.Entries) == 0 {
		return rf.LastIncludedTerm != rf.CurrentTerm
	}
	return rf.Entries[len(rf.Entries)-1].Term != rf.CurrentTerm
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//没加锁，所以只能在有锁环境调用
func (rf *Raft) persistData() []byte {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.CurrentTerm)
	e.Encode(rf.VotedFor)
	e.Encode(rf.Entries)
	e.Encode(rf.LastIncludedIndex)
	e.Encode(rf.LastIncludedTerm)

	data := w.Bytes()
	// 记录持久化日志
	rf.logger.Printf("记录持久化 Raft 状态: 当前任期=%d, 投票给=%d, 日志条目数=%d, 快照最后索引=%d, 快照最后任期=%d",
		rf.CurrentTerm,
		rf.VotedFor,
		len(rf.Entries),
		rf.LastIncludedIndex,
		rf.LastIncludedTerm)
	return data

}

func (rf *Raft) persist() {
	data := rf.persistData()
	rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}

	rf.mu.Lock()
	defer rf.mu.Unlock()

	// Your code here (2C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)

	var CurrentTerm int
	var VoteFor int
	var Entry []LogEntry
	var LastIncludedIndex int
	var LastIncludedTerm int
	//解码顺序必须和编码顺序一致
	if d.Decode(&CurrentTerm) != nil || d.Decode(&VoteFor) != nil || d.Decode(&Entry) != nil || d.Decode(&LastIncludedIndex) != nil || d.Decode(&LastIncludedTerm) != nil {
		rf.logger.Fatalf("Decode Error\n")
	} else {
		rf.CurrentTerm = CurrentTerm
		rf.VotedFor = VoteFor
		rf.Entries = Entry
		rf.LastIncludedIndex = LastIncludedIndex
		rf.LastIncludedTerm = LastIncludedTerm
		//快照相当于全部已应用给上层
		rf.lastApplied = rf.LastIncludedIndex
		rf.logger.Printf("读取持久化 Raft 状态: 当前任期=%d, 投票给=%d, 日志条目数=%d, 快照最后包含索引=%d, 快照最后包含任期=%d",
			rf.CurrentTerm,
			rf.VotedFor,
			len(rf.Entries),
			rf.LastIncludedIndex,
			rf.LastIncludedTerm)
	}
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

type InstallSnapshotArgs struct {
	LeaderId          int
	Term              int
	LastIncludedIndex int    // 快照对应的最后一条日志索引
	LastIncludedTerm  int    // 快照对应的最后一条日志任期
	Data              []byte // 快照数据
}

type InstallSnapshotReply struct {
	Term    int
	Success bool // 是否成功应用快照
}

//接收他人的快照
func (rf *Raft) RequestInstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	reply.Term = rf.CurrentTerm
	reply.Success = false

	//任期不对直接拒绝
	if rf.CurrentTerm > args.Term {
		rf.logger.Printf("[快照拒绝] 当前任期 %d 大于 Leader 任期 %d，拒绝安装快照",
			rf.CurrentTerm, args.Term)
		return
	}

	//转换为follower
	if rf.CurrentTerm < args.Term {
		oldState := rf.state
		oldTerm := rf.CurrentTerm
		rf.CurrentTerm = args.Term
		rf.VotedFor = -1
		rf.state = Follower
		//状态转换时（包括任期转换，投票，日志）都需要持久化
		rf.persist()

		rf.logger.Printf("[状态变更] 从 %s (任期 %d) 转换为 Follower (任期 %d)，因接收到 Leader %d 的快照,且任期更高",
			oldState, oldTerm, rf.CurrentTerm, args.LeaderId)
	} else if args.Term == rf.CurrentTerm && rf.state == Candidate {
		// 新增：当任期相同且当前是候选者状态时，转换为跟随者
		rf.logger.Printf("服务器 %d 在任期 %d 收到来自领导者 %d 的快宅，放弃竞选，转换为 Follower 状态",
			rf.me, rf.CurrentTerm, args.LeaderId)
		rf.state = Follower
		rf.VotedFor = -1
		// 状态转换时需要持久化
		rf.persist()
	}

	rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)

	//已经在其他部分并发提交了更高的快照
	if args.LastIncludedIndex <= rf.LastIncludedIndex {
		rf.logger.Printf("[快照忽略] 接收到的快照索引 %d 小于等于当前已安装的快照索引 %d",
			args.LastIncludedIndex, rf.LastIncludedIndex)
		return
	}

	reply.Success = true

	// 记录接收前的状态
	oldLastIncludedIndex := rf.LastIncludedIndex
	oldEntriesLen := len(rf.Entries)

	// 删除快照已覆盖的部分
	if args.LastIncludedIndex-rf.LastIncludedIndex < len(rf.Entries) {
		rf.Entries = rf.Entries[(args.LastIncludedIndex - rf.LastIncludedIndex):]
		rf.logger.Printf("[日志裁剪] 裁剪日志: 从索引 %d 到 %d，保留 %d 个日志条目",
			oldLastIncludedIndex, args.LastIncludedIndex, len(rf.Entries))
	} else {
		rf.Entries = make([]LogEntry, 0)
		rf.logger.Printf("[日志重置] 清空所有日志条目 (原长度: %d)，因快照覆盖了全部现有日志",
			oldEntriesLen)
	}

	// 更新并保存快照相关信息
	oldCommitIndex := rf.commitIndex
	oldLastApplied := rf.lastApplied
	rf.LastIncludedIndex = args.LastIncludedIndex
	rf.LastIncludedTerm = args.LastIncludedTerm

	// 更新提交索引
	if args.LastIncludedIndex > rf.commitIndex {
		rf.commitIndex = args.LastIncludedIndex
		rf.logger.Printf("[提交索引更新] 提交索引从 %d 更新为快照索引 %d",
			oldCommitIndex, rf.commitIndex)
	}

	// 更新最后应用索引
	if args.LastIncludedIndex > rf.lastApplied {
		rf.lastApplied = args.LastIncludedIndex
		rf.logger.Printf("[应用状态更新] 最后应用索引从 %d 更新为快照索引 %d",
			oldLastApplied, rf.lastApplied)
	}
	//rf.persist()
	//rf.persister.SaveSnapshot(args.Data)
	//改用原子操作
	rf.persister.SaveStateAndSnapshot(rf.persistData(), args.Data)
	// 异步应用快照（仅当快照包含新状态时）
	if args.LastIncludedIndex > oldLastApplied {
		rf.snapshotInProgress = args.LastIncludedIndex
		go func(startIndex int) {
			rf.logger.Printf("[快照应用] 向应用层发送快照应用消息 (索引: %d, 任期: %d, 大小: %d 字节)",
				args.LastIncludedIndex, args.LastIncludedTerm, len(args.Data))
			rf.applyChan <- ApplyMsg{
				SnapshotValid: true,
				SnapshotIndex: args.LastIncludedIndex,
				SnapshotTerm:  args.LastIncludedTerm,
				Snapshot:      args.Data,
			}
			rf.mu.Lock()
			//只释放自己加的，如果有更新的，不释放
			if rf.snapshotInProgress == startIndex {
				rf.snapshotInProgress = -1
				rf.snapshotCond.Broadcast() // 通知等待的日志
			}
			rf.mu.Unlock()
		}(rf.snapshotInProgress)
	}

	rf.logger.Printf("[快照接收成功] 成功接收并处理来自 Leader %d 的快照 (索引: %d, 任期: %d)",
		args.LeaderId, args.LastIncludedIndex, args.LastIncludedTerm)
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	//不能是未提交日志，或者已经存快照的日志
	if index > rf.commitIndex || index <= rf.LastIncludedIndex {
		rf.logger.Printf("快照操作失败: 索引=%d, 提交索引=%d, 最后包含索引=%d, 原因=索引超出范围",
			index, rf.commitIndex, rf.LastIncludedIndex)
		return
	}
	// 记录快照前的状态
	prevLastIncludedIndex := rf.LastIncludedIndex
	prevLogCount := len(rf.Entries)

	// 去掉index之前的，相当于存快照了
	rf.LastIncludedTerm = rf.Entries[index-prevLastIncludedIndex-1].Term
	rf.Entries = rf.Entries[index-prevLastIncludedIndex:]
	rf.LastIncludedIndex = index

	// 记录快照后的状态
	newLogCount := len(rf.Entries)
	snapshotSize := len(snapshot)

	//这一步基本不会走，因为这个是由服务器调用的，服务器根据我们提交的日志索引来的，我们提交的一定是applied的
	if rf.lastApplied < rf.LastIncludedIndex {
		rf.lastApplied = rf.LastIncludedIndex
	}

	rf.logger.Printf("创建快照成功: 快照最后索引=%d, 最后任期=%d, 快照前日志数=%d, 快照后日志数=%d, 快照大小=%.2fKB",
		index, rf.LastIncludedTerm, prevLogCount, newLogCount, float64(snapshotSize)/1024.0)

	//rf.persist()
	//rf.persister.SaveSnapshot(snapshot)
	//改用原子操作
	rf.persister.SaveStateAndSnapshot(rf.persistData(), snapshot)
}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	//2A
	//候选人的任期
	CandidateTerm int
	//候选人的Id
	CandidateId int
	//候选人的最新日志的索引
	LastLogIndex int
	//候选人最新日志的任期
	LastLogTerm int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	//投票人的任期
	VoterTerm int
	//是否投票
	VoteGranted bool
}

// 为ServerState添加String方法
func (s ServerState) String() string {
	switch s {
	case Leader:
		return "Leader"
	case Follower:
		return "Follower"
	case Candidate:
		return "Candidate"
	default:
		return fmt.Sprintf("Unknown(%d)", int(s))
	}
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	//2A
	rf.mu.Lock()
	defer rf.mu.Unlock()

	rf.logger.Printf("收到候选人 %d 的投票请求，任期为 %d，当前自身任期为 %d",
		args.CandidateId, args.CandidateTerm, rf.CurrentTerm)

	reply.VoterTerm = rf.CurrentTerm
	reply.VoteGranted = false

	// 请求者的任期小于当前任期拒绝
	if args.CandidateTerm < rf.CurrentTerm {
		rf.logger.Printf("拒绝投票给候选人 %d：请求任期 %d 小于当前任期 %d",
			args.CandidateId, args.CandidateTerm, rf.CurrentTerm)
		return
	}

	// 如果候选者的任期较大，转换为Follower
	if args.CandidateTerm > rf.CurrentTerm {
		oldState := rf.state
		oldTerm := rf.CurrentTerm
		rf.CurrentTerm = args.CandidateTerm
		rf.VotedFor = -1
		rf.state = Follower
		//状态转换时（包括任期转换，投票，日志）都需要持久化
		rf.persist()
		rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)

		rf.logger.Printf("状态变更：从 %s (任期 %d) 转换为 Follower (任期 %d)，因候选人 %d 任期更新",
			oldState, oldTerm, rf.CurrentTerm, args.CandidateId)
		//继续向下处理，需要投票给当前候选者
		//return
	}

	// 3. 快速路径：已投票给当前候选人，直接返回上次结果
	if rf.VotedFor == args.CandidateId {
		reply.VoteGranted = true                                                     // 已投票，直接返回成功
		rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond) // 重置超时器（确保活性）
		rf.logger.Printf("已投票给候选人 %d，直接同意投票请求，重置选举计时器", args.CandidateId)
		return
	}

	// 当前已经投票了且不是当前候选者，拒绝
	if rf.VotedFor != -1 {
		rf.logger.Printf("拒绝投票给候选人 %d：已投票给候选人 %d",
			args.CandidateId, rf.VotedFor)
		return
	}

	var LastLogTerm int
	var LastLogIndex int

	// 没有未快照的日志（Entries为空），最后一条日志是快照包含的lastIncludedIndex
	if len(rf.Entries) == 0 {
		LastLogIndex = rf.LastIncludedIndex
		LastLogTerm = rf.LastIncludedTerm
	} else {
		// 有未快照的日志，最后一条日志是Entries的最后一条
		// 实际索引 = lastIncludedIndex + 未快照日志的数量（len(Entries)）
		LastLogIndex = rf.LastIncludedIndex + len(rf.Entries)
		LastLogTerm = rf.Entries[len(rf.Entries)-1].Term
	}

	// 如果没有日志条目，全都是-1
	//LastLogIndex = len(rf.Entries)
	//LastLogTerm = 0
	//if LastLogIndex != 0 {
	//	LastLogTerm = rf.Entries[len(rf.Entries)-1].Term
	//}

	// 看看是不是一样新
	//旧的错误逻辑
	//if args.LastLogTerm > LastLogTerm {
	//	isUptoDate = true
	//} else {
	//	isUptoDate = args.LastLogIndex >= LastLogIndex
	//}
	// 修正日志比较逻辑
	var isUptoDate bool
	if args.LastLogTerm > LastLogTerm {
		isUptoDate = true
	} else if args.LastLogTerm < LastLogTerm {
		isUptoDate = false
	} else {
		// 任期相同，比较日志索引
		isUptoDate = args.LastLogIndex >= LastLogIndex
	}

	rf.logger.Printf("候选人 %d 日志状态检查：候选人最后日志 (term=%d, index=%d)，自身最后日志 (term=%d, index=%d)，是否更新：%v",
		args.CandidateId, args.LastLogTerm, args.LastLogIndex, LastLogTerm, LastLogIndex, isUptoDate)

	// 投票，投票不需要改变当前节点状态，由超时或者心跳或其他机制触发
	if isUptoDate {
		reply.VoteGranted = true
		rf.VotedFor = args.CandidateId
		//变更投票状态时需要持久化
		rf.persist()
		// 决定投票后需要重置超时器
		rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)

		rf.logger.Printf("投票给候选人 %d：候选人日志足够新，重置选举计时器", args.CandidateId)

		// 注意：这里是RequestVote处理函数，不会直接触发状态转换为Leader
		// 获取超过半数投票的逻辑通常在Candidate状态的处理函数中
	} else {
		rf.logger.Printf("拒绝投票给候选人 %d：候选人日志不够新", args.CandidateId)
	}

	return
}

//追加日志请求参数(心跳)
type AppendEntriesArgs struct {
	LeaderTerm int
	LeaderId   int
	//要同步的日志的上一条的index
	PrevLogIndex int
	//要同步的日志的上一条的term
	PrevLogTerm int
	//要同步的日志，不用指针是防止被其他raft修改
	Entries []LogEntry
	//Leader的最新commit的日志的index
	LeaderCommit int
}

//追加日志回复参数(心跳)
type AppendEntriesReply struct {
	FollowerTerm int
	//是否成功（接收日志或者心跳）
	Success bool
	//冲突的日志序号
	ConflictIndex int
	//冲突的日志任期，0表示是长度超长
	ConflictTerm int
}

func (rf *Raft) sendToCommitChan(cindex int) {
	rf.commitIndexChan <- cindex
}

//接收添加日志或者心跳请求
func (rf *Raft) RequestAppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	// 记录收到的心跳请求
	rf.logger.Printf("收到来自服务器 %d 的 AppendEntries 请求 [任期: %d, prevLogIndex: %d, prevLogTerm: %d]",
		args.LeaderId, args.LeaderTerm, args.PrevLogIndex, args.PrevLogTerm)

	reply.FollowerTerm = rf.CurrentTerm
	reply.Success = false
	//如果当前任期大于心跳任期，直接拒绝
	if args.LeaderTerm < rf.CurrentTerm {
		rf.logger.Printf("拒绝来自服务器 %d 的 AppendEntries 请求：领导者任期 %d 小于本地任期 %d",
			args.LeaderId, args.LeaderTerm, rf.CurrentTerm)
		return
	}
	// 如果领导者的任期较大，转换为Follower
	if args.LeaderTerm > rf.CurrentTerm {
		rf.logger.Printf("服务器 %d 检测到更高的任期 %d（本地任期 %d），转换为 Follower 状态",
			rf.me, args.LeaderTerm, rf.CurrentTerm)
		rf.CurrentTerm = args.LeaderTerm
		rf.VotedFor = -1
		rf.state = Follower
		//状态转换时（包括任期转换，投票，日志）都需要持久化
		rf.persist()
	} else if args.LeaderTerm == rf.CurrentTerm && rf.state == Candidate {
		// 新增：当任期相同且当前是候选者状态时，转换为跟随者
		rf.logger.Printf("服务器 %d 在任期 %d 收到来自领导者 %d 的心跳，放弃竞选，转换为 Follower 状态",
			rf.me, rf.CurrentTerm, args.LeaderId)
		rf.state = Follower
		rf.VotedFor = -1
		// 状态转换时需要持久化
		rf.persist()
	}
	// 重置选举超时，只要任期对的上，不用管日志是否对的上，都应该重置计时器，否则就开新选举了
	rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)

	//携带日志
	rf.logger.Printf("收到日志条目，前置日志索引: %d, 前置日志任期: %d", args.PrevLogIndex, args.PrevLogTerm)

	//判断是否携带日志
	valid := true
	// 计算当前节点的最大日志索引（快照最后索引 + 现有日志长度）
	maxLogIndex := rf.LastIncludedIndex + len(rf.Entries)
	rf.logger.Printf("当前节点日志范围：快照至 %d（任期 %d），未快照的Entries 共 %d 条（实际索引 %d 至 %d）",
		rf.LastIncludedIndex, rf.LastIncludedTerm, len(rf.Entries),
		rf.LastIncludedIndex+1, maxLogIndex)

	// 判断前置日志是否合法
	//发来的索引超了
	if args.PrevLogIndex > maxLogIndex {
		rf.logger.Printf("前置日志验证失败: 前置日志索引 %d 超出本地日志最大索引 %d",
			args.PrevLogIndex, maxLogIndex)

		valid = false
		//回复参数，现在加一，后期更新next索引时更加统一
		reply.ConflictIndex = maxLogIndex + 1
		reply.ConflictTerm = 0

	} else if args.PrevLogIndex < rf.LastIncludedIndex {
		// prevLogIndex 在快照范围内（已被截断）,直接设置为无效，要求从快照下一个开始
		rf.logger.Printf("前置日志验证失败: prevLogIndex %d 已被快照包含（快照至 %d），无法验证",
			args.PrevLogIndex, rf.LastIncludedIndex)

		valid = false
		// 告知Leader从快照后开始同步
		reply.ConflictIndex = rf.LastIncludedIndex + 1
		reply.ConflictTerm = 0

	} else if args.PrevLogIndex == rf.LastIncludedIndex { // 刚好等于快照长度，相当于原始版本的0的判断，此处需要额外判断任期是否相同
		// prevLogIndex 恰好是快照的最后索引，需校验任期是否匹配
		if args.PrevLogTerm != rf.LastIncludedTerm {
			rf.logger.Printf("前置日志验证失败: 快照最后索引 %d 任期 %d 与 prevLogTerm %d 不匹配",
				rf.LastIncludedIndex, rf.LastIncludedTerm, args.PrevLogTerm)

			//这种情况说明机制出错了，因为只有提交的才可以这样
			//panic("某一步出错")
			valid = false
			reply.ConflictTerm = rf.LastIncludedTerm
			reply.ConflictIndex = rf.LastIncludedIndex + 1

		} else {
			rf.logger.Printf("前置日志验证成功: prevLogIndex %d 与快照最后索引匹配（任期 %d）",
				args.PrevLogIndex, rf.LastIncludedTerm)
		}
	} else { //不是在快照内，正常方式对比
		//当前entries中的索引，不用考虑0
		entriesIndex := args.PrevLogIndex - rf.LastIncludedIndex - 1
		// 任期不匹配（冲突）
		if rf.Entries[entriesIndex].Term != args.PrevLogTerm {

			rf.logger.Printf("前置日志验证失败: prevLogIndex %d 任期 %d 与本地任期 %d 不匹配",
				args.PrevLogIndex, args.PrevLogTerm, rf.Entries[entriesIndex].Term)

			valid = false
			reply.ConflictTerm = rf.Entries[entriesIndex].Term
			reply.ConflictIndex = args.PrevLogIndex // 默认冲突点为prevLogIndex

			// 二分查找ConflictTerm在Entries中的最早索引（转换为实际索引）
			left, right := 0, entriesIndex

			for left <= right {
				mid := (left + right) / 2
				if rf.Entries[mid].Term == reply.ConflictTerm {
					reply.ConflictIndex = rf.LastIncludedIndex + 1 + mid
					right = mid - 1
				} else if rf.Entries[mid].Term < reply.ConflictTerm {
					left = mid + 1
				} else {
					right = mid - 1
				}
			}
		} else {
			rf.logger.Printf("前置日志验证成功: prevLogIndex %d 任期匹配（本地任期 %d）",
				args.PrevLogIndex, args.PrevLogTerm)
		}
	}

	var newEntriesAdded bool
	//如果日志合法或者是心跳请求
	if valid {
		//追加日志,找到冲突点之后再截断，再追加
		for i, entry := range args.Entries {
			//带快照的索引
			index := args.PrevLogIndex + i + 1
			//entries中的索引
			entriesIdx := index - rf.LastIncludedIndex - 1

			//如果是新的那种，就直接追加，主要是出现在冲突截断之后，直接添加，或者没有冲突，但是新发来的日志多
			if index > rf.LastIncludedIndex+len(rf.Entries) {
				newEntriesAdded = true
				rf.Entries = append(rf.Entries, entry)
				rf.logger.Printf("追加新日志: 实际索引 %d, 任期 %d, 命令 %d",
					index, entry.Term, entry.Command)
				continue
			}
			//日志冲突
			if rf.Entries[entriesIdx].Term != entry.Term {
				newEntriesAdded = true
				rf.logger.Printf("日志冲突: 本地索引 %d 任期为 %d，替换为任期 %d 的新条目",
					index, rf.Entries[entriesIdx].Term, entry.Term)

				// 记录截断前的日志状态
				oldLength := len(rf.Entries)

				// 截断
				rf.Entries = rf.Entries[:entriesIdx]
				rf.logger.Printf("截断日志: 从索引 %d 开始移除 %d 个条目，新日志长度为 %d",
					index, oldLength-len(rf.Entries), len(rf.Entries)+rf.LastIncludedIndex)

				rf.Entries = append(rf.Entries, entry)

				// 添加新的日志输出
				rf.logger.Printf("追加新日志: 索引 %d, 任期 %d, 命令 %v",
					index, entry.Term, entry.Command)
			}
		}
		if newEntriesAdded {
			//只有添加日志了才持久化
			rf.persist()
		}

		// 更新提交index
		oldCommit := rf.commitIndex
		if args.LeaderCommit > rf.commitIndex {
			cindex := 0
			if args.LeaderCommit > rf.LastIncludedIndex+len(rf.Entries) {
				rf.commitIndex = rf.LastIncludedIndex + len(rf.Entries)
				rf.logger.Printf("提交索引更新: 从 %d 更新至日志末尾 %d (领导者提交索引 %d 超出本地日志长度)",
					oldCommit, rf.commitIndex, args.LeaderCommit)
			} else {
				rf.commitIndex = args.LeaderCommit

				rf.logger.Printf("提交索引更新: 从 %d 更新至 %d (领导者提交索引 %d)",
					oldCommit, rf.commitIndex, args.LeaderCommit)
			}
			cindex = rf.commitIndex
			//更新索引，传递消息
			go rf.sendToCommitChan(cindex)
		}

		reply.Success = true
		rf.logger.Printf("成功响应服务器 %d 的 AppendEntries 请求，回复状态: Success=%v",
			args.LeaderId, reply.Success)
	} else {
		rf.logger.Printf("拒绝来自服务器 %d 的 AppendEntries 请求：前置日志不匹配", args.LeaderId)
	}
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	index := len(rf.Entries) + 1 + rf.LastIncludedIndex
	term := rf.CurrentTerm
	isLeader := rf.state == Leader

	if isLeader {
		newEntry := LogEntry{
			Term:    term,
			Index:   index,
			Command: command,
		}
		//添加但不立即发送，在心跳时统一发送处理
		rf.Entries = append(rf.Entries, newEntry)
		//更新领导者的matchindex
		rf.matchIndex[rf.me] = index
		rf.logger.Printf("领导者(任期 %d) 接收命令 %v，matchIndex（日志索引）更新为为 %d", term, command, index)
		//接收新命令时持久化
		go rf.sendHeartBeat(term)
		rf.persist()
	} else {
		rf.logger.Printf("非领导者(任期 %d) 拒绝命令 %v，当前状态：%s", term, command, rf.state)
	}

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.RequestAppendEntries", args, reply)
	return ok
}

func (rf *Raft) sendInstallSnapshot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) bool {
	ok := rf.peers[server].Call("Raft.RequestInstallSnapshot", args, reply)
	return ok
}

func (rf *Raft) updateCommitIndex() bool {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	// 确保commitIndex至少等于LastIncludedIndex（快照已提交）,理论不会出现这种情况
	if rf.commitIndex < rf.LastIncludedIndex {
		oldCommit := rf.commitIndex
		rf.commitIndex = rf.LastIncludedIndex
		rf.logger.Printf("提交索引同步至快照位置: 从%d到%d", oldCommit, rf.commitIndex)
		go rf.sendToCommitChan(rf.commitIndex)
	}

	matchIndex := make([]int, len(rf.peers))
	copy(matchIndex, rf.matchIndex)

	sort.Ints(matchIndex)
	midIndex := matchIndex[len(rf.peers)/2]

	oldCommitIndex := rf.commitIndex

	//只有更大才需要更新index
	if midIndex > oldCommitIndex {
		entriesIndex := midIndex - rf.LastIncludedIndex - 1
		entryTerm := rf.Entries[entriesIndex].Term
		if entryTerm == rf.CurrentTerm {
			rf.commitIndex = midIndex
			rf.logger.Printf("提交索引更新: 从%d到%d, 日志条目Term=%d",
				oldCommitIndex, midIndex, entryTerm)

			go rf.sendToCommitChan(midIndex)
			return true
		} else {
			rf.logger.Printf("提交索引未更新: 日志任期不匹配，日志任期=%d, 当前任期=%d",
				entryTerm, rf.CurrentTerm)
			return true
		}
	}

	rf.logger.Printf("提交索引未更新: 当前=%d, 中位数N=%d, 原因=条件不满足",
		oldCommitIndex, midIndex)
	return false
}

func (rf *Raft) handleCommitIndex() {
	latestUpdate := 0
	for rf.killed() == false {
		select {
		case cindex := <-rf.commitIndexChan:
			//已经并发传递过了
			if latestUpdate >= cindex {
				rf.logger.Printf("忽略过时的提交索引更新: 接收到cindex=%d, 但latestUpdate=%d", cindex, latestUpdate)
				continue
			}
			updated := rf.applyCommitMsg(cindex)
			if updated {
				latestUpdate = cindex
				rf.logger.Printf("提交索引更新成功: latestUpdate更新为%d", latestUpdate)
			} else {
				rf.logger.Printf("提交索引未更新: 可能已被其他协程处理")
			}
		}
	}
}

//向上层通道中准备消息
func (rf *Raft) applyCommitMsg(cindex int) bool {
	var msgs []ApplyMsg

	rf.mu.Lock()

	// 本次要提交的最大日志索引是 cindex
	// 等待条件：存在未完成的快照，且本次提交的日志中包含索引 > 快照索引的日志，那么必须等待快照被提交了才能提交
	// 即：快照索引 S 存在，且 cindex > S → 需要等待快照先应用
	for rf.snapshotInProgress != -1 && cindex > rf.snapshotInProgress {
		rf.logger.Printf("日志提交等待快照完成: 快照索引 %d, 待提交日志范围 [%d, %d]（包含>快照的日志）",
			rf.snapshotInProgress, rf.lastApplied+1, cindex)
		rf.snapshotCond.Wait() // 释放锁并等待快照完成
	}

	//代表其他的已经并发修改过了
	if cindex <= rf.lastApplied {
		rf.logger.Printf("放弃应用消息: 目标索引cindex=%d <= 当前lastApplied=%d", cindex, rf.lastApplied)
		rf.mu.Unlock()
		return false
	}
	oldLastApplied := rf.lastApplied

	// 提取要发送的消息分片
	for rf.lastApplied < cindex {
		rf.lastApplied++
		entry := rf.Entries[rf.lastApplied-rf.LastIncludedIndex-1]
		msg := ApplyMsg{
			CommandValid: true,
			Command:      entry.Command,
			CommandIndex: entry.Index,
			CommandTerm:  entry.Term,
		}
		msgs = append(msgs, msg)
	}

	newLastApplied := rf.lastApplied
	rf.mu.Unlock()

	// 记录日志
	if len(msgs) > 0 {
		rf.applyQueue <- msgs
		rf.logger.Printf("提交消息更新: 从 lastApplied=%d 到 %d, 提交消息数量=%d",
			oldLastApplied, newLastApplied, len(msgs))

	} else {
		rf.logger.Printf("提交消息未更新: 当前 lastApplied=%d, commitIndex=%d, 原因=无新消息",
			oldLastApplied, cindex)
	}
	return true
}

func (rf *Raft) submitApplyMsg() {
	for rf.killed() == false {
		select {
		case msgs := <-rf.applyQueue:

			firstIndex := msgs[0].CommandIndex
			lastIndex := msgs[len(msgs)-1].CommandIndex

			rf.logger.Printf("开始应用消息批次: 索引范围 [%d, %d], 数量=%d",
				firstIndex, lastIndex, len(msgs))

			for _, msg := range msgs {
				rf.applyChan <- msg
			}

			rf.logger.Printf("完成应用消息批次: 索引范围 [%d, %d]",
				firstIndex, lastIndex)
		}
	}
}

//找到并更新nextindex
func (rf *Raft) updateNextIndex(cindex, cterm, prevLogIndex, server int) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	// 记录进入函数时的参数信息
	rf.logger.Printf("进入 updateNextIndex - 服务器: %d, 冲突索引: %d, 冲突任期: %d, 前日志索引: %d",
		server, cindex, cterm, prevLogIndex)

	// 只有当冲突索引小于当前nextIndex时才需要更新
	if cindex >= rf.nextIndex[server] {
		rf.logger.Printf("服务器: %d, 冲突索引(%d)不小于当前nextIndex(%d)，无需更新", server, cindex, rf.nextIndex[server])
		return
	}

	// 意味着是prevLogIndex的长度超了
	if cterm == 0 {
		// 更新nextindex为冲突index+1
		rf.nextIndex[server] = cindex
		rf.logger.Printf("处理日志缺失冲突 - 服务器: %d, nextIndex 设置为: %d (冲突索引 %d + 1)",
			server, rf.nextIndex[server], cindex)
		return
	}

	// 二分查找找到cterm的最后一条日志
	rf.logger.Printf("开始二分查找任期 %d 的最后一个条目 - 搜索范围: [1, %d]", cterm, prevLogIndex)
	left := 0
	right := prevLogIndex - rf.LastIncludedIndex - 1
	lastIndex := -1

	for left <= right {
		mid := (left + right) / 2
		if rf.Entries[mid].Term == cterm {
			//index从1开始
			lastIndex = mid + 1
			left = mid + 1
		} else if rf.Entries[mid].Term < cterm {
			left = mid + 1
		} else {
			right = mid - 1
		}
	}

	// 找到相同任期的日志, 直接取最后一条的下一个作为nextindex
	if lastIndex != -1 {
		rf.nextIndex[server] = lastIndex + 1 + rf.LastIncludedIndex
		rf.logger.Printf("找到任期 %d 的最后一个条目在索引 %d，nextIndex 设置为: %d",
			cterm, lastIndex, rf.nextIndex[server])
	} else { // 没找到 说明当前term在leader中不存在，要全都清除
		rf.nextIndex[server] = cindex
		rf.logger.Printf("未找到任期 %d 的条目，nextIndex 设置为冲突索引: %d", cterm, rf.nextIndex[server])
	}
}

func (rf *Raft) sendHeartBeat(term int) {
	// 获取初始信息
	rf.mu.RLock()
	// 快速检查状态，避免不必要的操作
	if term != rf.CurrentTerm || rf.state != Leader {
		if term != rf.CurrentTerm {
			rf.logger.Printf("取消心跳发送：任期已从 %d 变更为 %d", term, rf.CurrentTerm)
		} else {
			rf.logger.Printf("取消心跳发送：状态已经已从 Leader 变更为 %s", rf.state)
		}
		rf.mu.RUnlock()
		return
	}

	// 保存必要的状态信息，减少锁的持有时间
	leaderCommit := rf.commitIndex
	nextIndexs := make([]int, len(rf.nextIndex))
	copy(nextIndexs, rf.nextIndex)
	entries := make([]LogEntry, len(rf.Entries))
	copy(entries, rf.Entries)

	lastIndex := rf.LastIncludedIndex
	lastTerm := rf.LastIncludedTerm

	data := rf.persister.ReadSnapshot()

	rf.mu.RUnlock()

	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}
		go func(server int, nextIndex int) {

			//发送心跳前先检查是否需要发送快照
			if nextIndex <= lastIndex {
				snapshotArgs := InstallSnapshotArgs{
					LeaderId:          rf.me,
					Term:              term,
					LastIncludedIndex: lastIndex,
					LastIncludedTerm:  lastTerm,
					Data:              data,
				}
				snapshotReply := InstallSnapshotReply{}

				rf.logger.Printf("Leader（ID: %d，任期: %d）开始向节点%d发送快照，快照信息 index %d，term %d",
					rf.me, term, server, lastIndex, lastTerm)

				//如果发送快照就暂停其他日志同步
				if rf.sendInstallSnapshot(server, &snapshotArgs, &snapshotReply) {
					rf.mu.Lock()
					defer rf.mu.Unlock()

					currentTerm := rf.CurrentTerm
					// 提交心跳时的任期和现在的任期不同，说明阶段发生变化要忽略之前的请求回复
					if term != currentTerm {
						rf.logger.Printf("忽略对服务器 %d 的快照回复：任期已从 %d 变更为 %d", server, term, currentTerm)
						return
					}
					// 如果不再是领导者，忽略回复
					if rf.state != Leader {
						rf.logger.Printf("忽略对服务器 %d 的快照回复：本节点状态已从 Leader 变更为 %s", server, rf.state)
						return
					}

					// 追随者的任期更高，直接退出，转换为 follower
					if term < snapshotReply.Term {
						rf.CurrentTerm = snapshotReply.Term
						rf.VotedFor = -1
						rf.state = Follower
						rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)
						//状态转换时（包括任期转换，投票，日志）都需要持久化
						rf.persist()
						rf.logger.Printf("转换为 Follower（任期: %d）：服务器 %d 的任期 %d 更高", snapshotReply.Term, server, snapshotReply.Term)
						return
					}
					if !snapshotReply.Success {
						rf.logger.Printf("服务器 %d 拒绝接收快照：当前快照可能已经过时或者被接收", server)
						//如果消息丢失导致nextindex没有更新，再检查一次
						if lastIndex+1 > rf.nextIndex[server] {
							oldNextIndex := rf.nextIndex[server]
							oldMatchIndex := rf.matchIndex[server]

							rf.nextIndex[server] = lastIndex + 1
							rf.matchIndex[server] = lastIndex

							rf.logger.Printf("发送快照失败，但更新服务器 %d 的索引 - nextIndex 从 %d → %d，matchIndex 从 %d → %d",
								server, oldNextIndex, rf.nextIndex[server], oldMatchIndex, rf.matchIndex[server])
						}
						return
					}

					rf.logger.Printf("服务器 %d 接收快照：准备开始更新索引", server)
					//更新nextIndex和commitIndex
					oldMatchIndex := rf.matchIndex[server]
					oldNextIndex := rf.nextIndex[server]

					// 更新nextIndex和matchIndex
					newMatchIndex := lastIndex
					// 只有当新值大于旧值时才更新
					if newMatchIndex > oldMatchIndex {
						rf.matchIndex[server] = newMatchIndex
						rf.nextIndex[server] = newMatchIndex + 1

						newNextIndex := rf.nextIndex[server]
						// 更新索引
						go rf.updateCommitIndex()

						rf.logger.Printf("服务器 %d（任期: %d）快照更新成功：matchIndex 从 %d 更新为 %d，nextIndex 从 %d 更新为 %d",
							server, term, oldMatchIndex, newMatchIndex, oldNextIndex, newNextIndex)
					} else {
						rf.logger.Printf("服务器 %d（任期: %d）日志状态无需更新：matchIndex 保持为 %d，nextIndex 保持为 %d",
							server, term, oldMatchIndex, oldNextIndex)
					}
				} else {
					rf.logger.Printf("给节点%d发送快照请求失败", server)
				}
				return
			}

			// 计算要发送的日志条目数量和起始索引
			// 计算要发送的日志条目
			prevLogIndex := nextIndex - 1
			prevLogTerm := lastTerm
			//前任日志不是快照的最后一个
			if prevLogIndex > lastIndex {
				prevLogTerm = entries[prevLogIndex-lastIndex-1].Term
			}
			//else {
			//	prevLogIndex = lastIndex
			//}
			// 截取要发送的日志条目
			var entriesSend []LogEntry
			if (prevLogIndex - lastIndex) < len(entries) {
				entriesSend = entries[(prevLogIndex - lastIndex):]
			} else {
				entriesSend = []LogEntry{}
			}

			rf.logger.Printf("Leader（ID: %d，任期: %d）开始向节点%d发送AppendEntries，最新日志 (term=%d, index=%d)，发送日志条目数量=%d",
				rf.me, term, server, prevLogTerm, prevLogIndex, len(entriesSend))

			args := AppendEntriesArgs{
				LeaderTerm:   term,
				LeaderId:     rf.me,
				PrevLogIndex: prevLogIndex,
				PrevLogTerm:  prevLogTerm,
				Entries:      entriesSend,
				LeaderCommit: leaderCommit,
			}
			reply := AppendEntriesReply{}
			// 发送心跳请求
			if rf.sendAppendEntries(server, &args, &reply) {
				rf.mu.Lock()
				defer rf.mu.Unlock()
				currentTerm := rf.CurrentTerm
				// 提交心跳时的任期和现在的任期不同，说明阶段发生变化要忽略之前的请求回复
				if term != currentTerm {
					rf.logger.Printf("忽略对服务器 %d 的心跳回复：任期已从 %d 变更为 %d", server, term, currentTerm)
					return
				}
				// 如果不再是领导者，忽略回复
				if rf.state != Leader {
					rf.logger.Printf("忽略对服务器 %d 的心跳回复：本节点状态已从 Leader 变更为 %s", server, rf.state)
					return
				}

				// 追随者的任期更高，直接退出，转换为 follower
				if term < reply.FollowerTerm {
					rf.CurrentTerm = reply.FollowerTerm
					rf.VotedFor = -1
					rf.state = Follower
					rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)
					//状态转换时（包括任期转换，投票，日志）都需要持久化
					rf.persist()
					rf.logger.Printf("转换为 Follower（任期: %d）：服务器 %d 的任期 %d 更高", reply.FollowerTerm, server, reply.FollowerTerm)
					return
				}
				// 成功收到追随者响应
				if reply.Success {
					rf.logger.Printf("服务器 %d 成功接收心跳和日志同步（任期: %d）", server, term)
					oldMatchIndex := rf.matchIndex[server]
					oldNextIndex := rf.nextIndex[server]

					// 更新nextIndex和matchIndex
					newMatchIndex := args.PrevLogIndex + len(args.Entries)
					// 只有当新值大于旧值时才更新
					if newMatchIndex > oldMatchIndex {
						rf.matchIndex[server] = newMatchIndex
						rf.nextIndex[server] = newMatchIndex + 1

						newNextIndex := rf.nextIndex[server]
						// 更新索引
						go rf.updateCommitIndex()

						rf.logger.Printf("服务器 %d（任期: %d）日志复制成功：matchIndex 从 %d 更新为 %d，nextIndex 从 %d 更新为 %d",
							server, term, oldMatchIndex, newMatchIndex, oldNextIndex, newNextIndex)
					} else {
						rf.logger.Printf("服务器 %d（任期: %d）日志状态无需更新：matchIndex 保持为 %d，nextIndex 保持为 %d",
							server, term, oldMatchIndex, oldNextIndex)
					}
				} else {
					// 失败：回退nextIndex 用二分法查找
					//根据冲突信息更新nextindex
					localServer := server
					go rf.updateNextIndex(reply.ConflictIndex, reply.ConflictTerm, args.PrevLogIndex, localServer)
				}
			} else {
				rf.logger.Printf("向服务器 %d 发送心跳失败（任期: %d）", server, term)
			}
		}(i, nextIndexs[i])
	}
}

//发起投票
func (rf *Raft) startVote(term int) {
	rf.mu.RLock()
	// 快速检查状态，避免不必要的操作
	if term != rf.CurrentTerm || rf.state != Candidate {
		if term != rf.CurrentTerm {
			rf.logger.Printf("取消发送投票：任期已从 %d 变更为 %d", term, rf.CurrentTerm)
		} else {
			rf.logger.Printf("取消发送投票：状态已经已从 Candidate 变更为 %s", rf.state)
		}
		rf.mu.RUnlock()
		return
	}

	LastLogIndex := len(rf.Entries) + rf.LastIncludedIndex
	LastLogTerm := rf.LastIncludedTerm
	if LastLogIndex > rf.LastIncludedIndex {
		LastLogTerm = rf.Entries[LastLogIndex-rf.LastIncludedIndex-1].Term
	}

	rf.mu.RUnlock()
	// 需要多少投票
	vNeeded := int32(len(rf.peers)/2 + 1)
	// 当前投票给本节点的数目，初始一票为自己投的
	var vCount int32
	atomic.StoreInt32(&vCount, 1)
	// 判断当前节点状态是否已经发生变化，任期，state，变化了就忽略本次回复
	var statusChanged int32

	rf.logger.Printf("节点 %d 在任期 %d 发起投票，需要 %d 票才能成为领导者", rf.me, term, vNeeded)

	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}
		go func(server int) {
			args := RequestVoteArgs{
				CandidateTerm: term,
				CandidateId:   rf.me,
				LastLogIndex:  LastLogIndex,
				LastLogTerm:   LastLogTerm,
			}
			reply := RequestVoteReply{}
			// 发送投票请求
			rf.logger.Printf("向服务器 %d 发送投票请求（任期: %d）", server, term)
			if rf.sendRequestVote(server, &args, &reply) {
				// 如果已经是领导了就不用再获取锁了，然后解锁了
				if atomic.LoadInt32(&statusChanged) == 1 {
					rf.logger.Printf("忽略服务器 %d 的投票回复：本节点状态已经发生变化，旧节点信息：（任期: %d）", server, term)
					return
				}
				rf.mu.Lock()
				defer rf.mu.Unlock()
				// 提交投票时的任期和现在的任期不同，说明阶段发生变化要忽略之前的请求回复
				currentTerm := rf.CurrentTerm
				if term != currentTerm {
					rf.logger.Printf("忽略服务器 %d 的投票回复：任期已从 %d 变更为 %d", server, term, currentTerm)
					atomic.StoreInt32(&statusChanged, 1)
					return
				}
				// 检查状态是否仍然是 Candidate
				if rf.state != Candidate {
					rf.logger.Printf("忽略服务器 %d 的投票回复：本节点状态已从 Candidate 变更为 %s（任期: %d）",
						server, rf.state, term)
					return
				}
				// 投票人的任期更高，直接退出，转换为 follower
				if term < reply.VoterTerm {
					atomic.StoreInt32(&statusChanged, 1)
					oldState := rf.state
					oldTerm := rf.CurrentTerm
					rf.CurrentTerm = reply.VoterTerm
					rf.VotedFor = -1
					rf.state = Follower
					rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)
					//状态转换时（包括任期转换，投票，日志）都需要持久化
					rf.persist()
					rf.logger.Printf("状态变更：从 %s (任期 %d) 转换为 Follower (任期 %d)，因服务器 %d 任期更新",
						oldState, oldTerm, reply.VoterTerm, server)
					return
				}
				// 获得投票
				if reply.VoteGranted {
					atomic.AddInt32(&vCount, 1)
					rf.logger.Printf("从服务器 %d 获得投票（任期: %d），当前票数: %d", server, term, atomic.LoadInt32(&vCount))
					// 超过了所需票数
					if atomic.LoadInt32(&vCount) >= vNeeded {
						if rf.state == Leader {
							rf.logger.Printf("忽略服务器 %d 的投票回复：本节点已成为领导者（任期: %d）", server, term)
							return
						}
						if rf.state != Candidate || rf.CurrentTerm != term {
							rf.logger.Printf("忽略服务器 %d 的投票回复：本节点状态已经发生变化，旧节点信息：（任期: %d）", server, term)
							atomic.StoreInt32(&statusChanged, 1)
							return
						}
						oldState := rf.state
						oldTerm := rf.CurrentTerm
						rf.state = Leader
						//leader的投票没必要重置为-1，主要是用于在投票期间确保只给一个人
						//如果领导者主动将VotedFor重置为 - 1，它可能在任期内再次投票给其他候选者，
						// 导致同一任期内出现多个领导者
						//rf.VotedFor = -1
						rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)
						//重置index信息
						rf.nextIndex = make([]int, len(rf.peers))
						for i := range rf.nextIndex {
							rf.nextIndex[i] = len(rf.Entries) + rf.LastIncludedIndex + 1
						}
						rf.matchIndex = make([]int, len(rf.peers))
						//领导者自己的应该初始化为entries的长度 + 已快照的长度
						rf.matchIndex[rf.me] = len(rf.Entries) + rf.LastIncludedIndex
						rf.logger.Printf("获取超过半数投票,票数为%d，从 %s (任期 %d) 转换为 Leader",
							atomic.LoadInt32(&vCount), oldState, oldTerm)
						// 成为 leader 时立即触发心跳
						rf.logger.Printf("即将发送初始心跳（ID: %d，任期: %d）", rf.me, oldTerm)
						go rf.sendHeartBeat(oldTerm)
						// 原子修改
						atomic.StoreInt32(&statusChanged, 1)
					}
				} else {
					rf.logger.Printf("被服务器 %d 拒绝投票（任期: %d），可能是日志不够新或者已经投给其他人，当前票数: %d", server, term, atomic.LoadInt32(&vCount))
				}
			} else {
				rf.logger.Printf("向服务器 %d 发送投票请求失败（任期: %d）", server, term)
			}
		}(i)
	}
}

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ticker() {
	for rf.killed() == false {

		// Your code here to check if a leader election should
		// be started and to randomize sleeping time using
		// time.Sleep().
		select {
		//定时选举任务
		case <-rf.electionTimer.C:
			rf.mu.Lock()
			//重置timer
			rf.electionTimer.Reset(time.Duration(200+rand.Intn(200)) * time.Millisecond)
			//follwer考虑收到心跳超时，candidate考虑选举超时
			if rf.state != Leader {
				//转换为候选者
				rf.state = Candidate
				//任期加一
				rf.CurrentTerm++
				term := rf.CurrentTerm
				//给自己投票
				rf.VotedFor = rf.me
				//开启选举时持久化
				rf.persist()
				rf.logger.Printf("等待心跳或竞选超时，重置计时器，转换为Candidate，当前任期为%d", rf.CurrentTerm)
				go rf.startVote(term)
			}
			rf.mu.Unlock()
		case <-rf.heartBeatTimer.C:
			rf.mu.Lock()
			//timer都是一次性的，需要重置
			rf.heartBeatTimer.Reset(100 * time.Millisecond)
			curState := rf.state
			term := rf.CurrentTerm
			rf.mu.Unlock()
			//只有Leader才处理心跳信号
			if curState == Leader {
				go rf.sendHeartBeat(term)
			}

		}
	}
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//

func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	// 设置日志前缀为"raft[me]:"，格式包含时间戳
	//rf.logger = log.New(os.Stdout, fmt.Sprintf("raft[%d]: ", me), log.LstdFlags)
	//rf.logger = log.New(os.Stdout, fmt.Sprintf("raft[%d]: ", me), log.LstdFlags|log.Lmicroseconds)
	rf.logger = log.New(io.Discard, fmt.Sprintf("raft[%d]: ", me), log.LstdFlags|log.Lmicroseconds)

	//初始化随机数种子
	rand.Seed(time.Now().UnixNano())
	//2A
	rf.CurrentTerm = 0
	rf.VotedFor = -1
	rf.dead = 0
	rf.electionTimer = time.NewTimer(time.Duration(200+rand.Intn(200)) * time.Millisecond)
	rf.heartBeatTimer = time.NewTimer(time.Duration(100) * time.Millisecond)
	rf.state = Follower
	rf.Entries = make([]LogEntry, 0)
	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	//2B
	rf.nextIndex = make([]int, len(peers))
	for i := range rf.nextIndex {
		rf.nextIndex[i] = 1 // 初始化为1，表示从第一个日志条目开始
	}
	rf.matchIndex = make([]int, len(peers))
	//发送消息的通道
	rf.applyChan = applyCh
	// 初始化无缓冲通道
	rf.applyQueue = make(chan []ApplyMsg)
	rf.commitIndexChan = make(chan int)
	rf.lastApplied = 0

	rf.snapshotInProgress = -1
	rf.snapshotCond = sync.NewCond(&rf.mu)

	//2c
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()
	//持续处理commitindex
	go rf.handleCommitIndex()
	//向上层发送消息
	go rf.submitApplyMsg()

	return rf
}
